package com.wudl.flink.hdfs;

import com.wudl.flink.hdfs.sink.HdfsSink;
import com.wudl.flink.hdfs.source.MySource;
import org.apache.commons.lang3.SystemUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * @author ：wudl
 * @date ：Created in 2021-12-27 0:12
 * @description：
 * @modified By：
 * @version: 1.0
 */

public class Application {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://192.168.1.161:8020/flink-checkpoint/checkpoint");
        //设置两个Checkpoint 之间最少等待时间,如设置Checkpoint之间最少是要等 500ms(为了避免每隔1000ms做一次Checkpoint的时候,前一次太慢和后一次重叠到一起去了  --//默认是0
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //设置Checkpoint的时间间隔为1000ms做一次Checkpoint/其实就是每隔1000ms发一次Barrier!
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //设置checkpoint的超时时间,如果 Checkpoint在 60s内尚未完成说明该次Checkpoint失败,则丢弃。
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        //设置同一时间有多少个checkpoint可以同时执行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        // 设置重启策略
        // 一个时间段内的最大失败次数
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3,
                // 衡量失败次数的是时间段
                Time.of(5, TimeUnit.MINUTES),
                // 间隔
                Time.of(10, TimeUnit.SECONDS)
        ));


        DataStreamSource<String> mySouceData = env.addSource(new MySource());
        FileSink<String> hdfsSink = HdfsSink.getHdfsSink();

        mySouceData.sinkTo(hdfsSink );
        mySouceData.print();


        env.execute();

    }
}
